package com.atguigu.gmall.app.function;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.util.DimUtil;
import com.atguigu.gmall.util.DruidDSUtils;
import com.atguigu.gmall.util.ThreadPoolUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.util.Collections;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @author : ranzlupup
 * @since : 2023/6/6 12:01
 */
public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T, T> implements DimJoinFunction<T>{
    private ThreadPoolExecutor threadPoolExecutor = null;
    private DruidDataSource dataSource = null;

    private String tableName;

    public DimAsyncFunction() {
    }

    public DimAsyncFunction(String tableName) {
        this.tableName = tableName;
    }

    // /**
    //  * 抽象方法，调用方法时重写getKey，动态传入查询维度表的key
    //  */
    // public abstract String getKey(T input);
    //
    // /**
    //  * 抽象方法，调用方法时重写join，使用查询到的维表数据补充input中的维表信息
    //  */
    // public abstract void join(T input, JSONObject dimInfo);


    @Override
    public void open(Configuration parameters) throws Exception {
        dataSource = DruidDSUtils.createDataSource();
        threadPoolExecutor = ThreadPoolUtil.getThreadPoolExecutor();
    }

    /**
     * @param input        element coming from an upstream task 来自上游任务的元素
     * @param resultFuture to be completed with the result data 结果数据
     */
    @Override
    public void asyncInvoke(T input, ResultFuture<T> resultFuture) throws Exception {
        threadPoolExecutor.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    // 获取连接
                    DruidPooledConnection connection = dataSource.getConnection();

                    // 查询维度表获取维度信息
                    String key = getKey(input);
                    JSONObject dimInfo = DimUtil.getDimInfo(connection, tableName, key);

                    // 将维度信息补充至当前数据
                    if (dimInfo != null) {
                        join(input, dimInfo);
                    }

                    // 归还连接
                    connection.close();

                    // 将结果写出
                    resultFuture.complete(Collections.singletonList(input));
                } catch (Exception e) {
                    e.printStackTrace();
                    System.out.println("关联维表失败：" + input + ",Table:" + tableName);
                    // resultFuture.complete(Collections.singletonList(input));
                }
            }
        });

    }


    @Override
    public void timeout(T input, ResultFuture<T> resultFuture) throws Exception {
        System.out.println("TimeOut:" + input);
    }
}
